package com;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.*;

/**
 *  Kafka 生产者
 */
public class KafkaProduct extends Thread{

    private String topic;
    private KafkaProducer<String,String> producer;
    public KafkaProduct(String topic){
        this.topic = topic;
        Properties props = new Properties();
        props.put("bootstrap.servers",KafkaPro.BROKER_LIST);
        props.put("acks","all");
        //重试次数
        props.put("retries",1);
        //批次大小
        props.put("batch.size",16384);
        //等待时间
        props.put("linger.ms",1);
        props.put("buffer.memory",33554432);
        props.put("key.serializer",   "org.apache.kafka.common.serialization.StringSerializer");
        // 使用该类将值对象序列化为字节数组
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        producer = new KafkaProducer(props);

    }

    @Override
    public void run() {
        Random random = new Random();
        //班级号
        List<String> listClass = Arrays.asList("202112","202111");
        //姓名
        List<String> listName = Arrays.asList("Jack","Tom","Ahern","Alonzo","Hunter","Vincent","Mackenzie","Mads","Frederick","Fabiana","Quentin");
        //性别
        List<String> listSex = Arrays.asList("1","2");
        //出生年月日
        List<String> listBirth = Arrays.asList("2000-10-10","2000-10-16");
        //学期
        List<String> listTerm = Arrays.asList("1","2","3","4","5","6","7","8");
        //绩点
        List<String> listGpa = Arrays.asList("1","2","3","4","5","6","7","8","9","10");
        //在籍状态
        List<String> listState = Arrays.asList("L","A");
        while (true){
            //班级号 姓名 性别 出生年月日 学号 学期 在籍状态
            Collections.shuffle(listClass);
            Collections.shuffle(listName);
            Collections.shuffle(listSex);
            Collections.shuffle(listBirth);
            String stuNum = String.valueOf(200000 + random.nextInt(100));
            Collections.shuffle(listTerm);
            Collections.shuffle(listGpa);
            Collections.shuffle(listState);
            String message = listClass.get(0)+"\t"+listName.get(0)+"\t"+listSex.get(0)+"\t"+listBirth.get(0)+"\t"+ stuNum +"\t"+listTerm.get(0)+"\t"+listGpa.get(0)+"\t"+listState.get(0);
            ProducerRecord pr = new ProducerRecord(topic,message);
            producer.send(pr);
            System.out.println("Send: "+message);
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

